package com.beifeng.hadoop.hdfs.secondarysort;

import java.io.IOException;  
import java.util.StringTokenizer;  

import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;  
import org.apache.hadoop.io.IntWritable;  
import org.apache.hadoop.io.LongWritable;  
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.mapreduce.Job;  
import org.apache.hadoop.mapreduce.Mapper;  
import org.apache.hadoop.mapreduce.Reducer;  
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  

  
public class SecondarySort {        
    // 自定义map  
    public static class Map extends  
            Mapper<LongWritable, Text, IntPair, IntWritable> {  
        private final IntPair outkey = new IntPair();  
        private final IntWritable outvalue = new IntWritable();  
        public void map(LongWritable key, Text value, Context context)  
                throws IOException, InterruptedException {  
            String line = value.toString();  
            StringTokenizer tokenizer = new StringTokenizer(line);  
            int left = 0;  
            int right = 0;  
            if (tokenizer.hasMoreTokens()) {  
                left = Integer.parseInt(tokenizer.nextToken());  
                if (tokenizer.hasMoreTokens())  
                    right = Integer.parseInt(tokenizer.nextToken());  
                outkey.set(left, right);  
                outvalue.set(right);  
                context.write(outkey, outvalue);  
            }  
        }  
    }  
    // 自定义reduce  
    //  
    public static class Reduce extends  
            Reducer<IntPair, IntWritable, Text, IntWritable> {  
        private Text left = new Text();  
         private static final Text LINE_TEXT= new Text("*********************"); 
        
        public void reduce(IntPair key, Iterable<IntWritable> values,  
                Context context) throws IOException, InterruptedException { 
        	
            context.write(LINE_TEXT, null);  
            left.set(Integer.toString(key.getFirst()));  
            
            for (IntWritable val : values) {  
            	
                context.write(left, val);  
            }  
        }  
    }  
    
    
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {  
        
    	 		//1.获取hadoop的默认的配置信息
    			Configuration conf = new Configuration();
    			
    			//2.生成对应的job
    			Job job = Job.getInstance(conf, "secondarysort");
    			
    			//设置提交jar到集群上运行，如果不写，会报错
    			job.setJarByClass(SecondarySort.class);
    			
    			//3.设置job的内容
    			
    			//3.1 输入路径
    			FileInputFormat.setInputPaths(job, "hdfs://bigdata01:8020/second.txt");
    			
    			//3.2 map
    			job.setMapperClass(Map.class);
    			job.setMapOutputKeyClass(IntPair.class);
    			job.setMapOutputValueClass(IntWritable.class);
    			
    			//===========shuffle=========
    			
    			job.setGroupingComparatorClass(GoupingCompartor.class);
    			job.setPartitionerClass(FirstPartitioner.class);
    			   			
    			//===========shuffle=========
    			//3.3 reduce
    			job.setReducerClass(Reduce.class);
    			job.setOutputKeyClass(Text.class);
    			job.setOutputValueClass(IntWritable.class);
   			    			
    			//3.4 输出路径
    			Path outpath = new Path("hdfs://bigdata01:8020/output_second");
    			//输出目录如果存在的话就把它删除
    			FileSystem  fsh = outpath.getFileSystem(conf);
    			if (fsh.exists(outpath)) {
    				fsh.delete(outpath, true);
    			}
    			
    			FileOutputFormat.setOutputPath(job, outpath);
    			
    			//4 提交的job是否运行成功

          		FileOutputFormat.setOutputPath(job, outpath);
            // 提交job  
            System.exit(job.waitForCompletion(true) ? 0 : 1);  
        
    }  
}  
